package com.example.flinkcourse.lesson4.v2;

import com.example.flinkcourse.lesson4.model.Event;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * DataStream API V2 处理器
 * 使用 V2 版本的 API 实现数据处理功能
 */
public class DataStreamV2Processor {
    private static final Logger LOG = LoggerFactory.getLogger(DataStreamV2Processor.class);

    /**
     * 处理事件流
     * @param inputStream 输入事件流
     * @return 处理后的事件流
     */
    public static DataStream<Event> processStream(DataStream<Event> inputStream) {
        LOG.info("Processing stream with DataStream API V2...");
        
        return inputStream
            .keyBy(Event::getType)
            .process(new KeyedProcessFunction<String, Event, Event>() {
                private ValueState<Long> countState;
                private ValueState<Double> sumState;
                
                @Override
                public void open(Configuration parameters) {
                    LOG.info("Initializing state...");
                    
                    // 初始化计数状态
                    ValueStateDescriptor<Long> countDescriptor = 
                        new ValueStateDescriptor<>("count", Types.LONG);
                    countState = getRuntimeContext().getState(countDescriptor);
                    
                    // 初始化求和状态
                    ValueStateDescriptor<Double> sumDescriptor = 
                        new ValueStateDescriptor<>("sum", Types.DOUBLE);
                    sumState = getRuntimeContext().getState(sumDescriptor);
                }
                
                @Override
                public void processElement(Event event, Context ctx, Collector<Event> out) {
                    try {
                        LOG.debug("Processing event: {}", event.getId());
                        
                        // 更新计数状态
                        Long count = countState.value();
                        if (count == null) {
                            count = 0L;
                        }
                        count++;
                        countState.update(count);
                        
                        // 更新求和状态
                        Double sum = sumState.value();
                        if (sum == null) {
                            sum = 0.0;
                        }
                        sum += event.getAmount();
                        sumState.update(sum);
                        
                        // 更新事件
                        event.setCount(count);
                        event.getProperties().put("sum", String.valueOf(sum));
                        
                        // 输出事件
                        out.collect(event);
                        
                        LOG.debug("Event processed successfully: {}", event.getId());
                    } catch (Exception e) {
                        LOG.error("Error processing event: " + event.getId(), e);
                    }
                }
            })
            .setParallelism(4);
    }
    
    /**
     * 处理带时间窗口的事件流
     * @param inputStream 输入事件流
     * @return 处理后的事件流
     */
    public static DataStream<Event> processStreamWithWindow(DataStream<Event> inputStream) {
        LOG.info("Processing stream with window using DataStream API V2...");
        
        return inputStream
            .keyBy(Event::getType)
            .process(new KeyedProcessFunction<String, Event, Event>() {
                private ValueState<Long> countState;
                private ValueState<Double> sumState;
                private ValueState<Long> windowStartState;
                
                @Override
                public void open(Configuration parameters) {
                    LOG.info("Initializing window state...");
                    
                    // 初始化计数状态
                    ValueStateDescriptor<Long> countDescriptor = 
                        new ValueStateDescriptor<>("count", Types.LONG);
                    countState = getRuntimeContext().getState(countDescriptor);
                    
                    // 初始化求和状态
                    ValueStateDescriptor<Double> sumDescriptor = 
                        new ValueStateDescriptor<>("sum", Types.DOUBLE);
                    sumState = getRuntimeContext().getState(sumDescriptor);
                    
                    // 初始化窗口开始时间状态
                    ValueStateDescriptor<Long> windowStartDescriptor = 
                        new ValueStateDescriptor<>("window-start", Types.LONG);
                    windowStartState = getRuntimeContext().getState(windowStartDescriptor);
                }
                
                @Override
                public void processElement(Event event, Context ctx, Collector<Event> out) {
                    try {
                        LOG.debug("Processing event in window: {}", event.getId());
                        
                        // 获取当前窗口开始时间
                        Long windowStart = windowStartState.value();
                        if (windowStart == null) {
                            windowStart = ctx.timestamp() - (ctx.timestamp() % 60000); // 1分钟窗口
                            windowStartState.update(windowStart);
                        }
                        
                        // 检查是否需要重置窗口
                        if (ctx.timestamp() >= windowStart + 60000) {
                            // 重置状态
                            countState.clear();
                            sumState.clear();
                            windowStart = ctx.timestamp() - (ctx.timestamp() % 60000);
                            windowStartState.update(windowStart);
                        }
                        
                        // 更新计数状态
                        Long count = countState.value();
                        if (count == null) {
                            count = 0L;
                        }
                        count++;
                        countState.update(count);
                        
                        // 更新求和状态
                        Double sum = sumState.value();
                        if (sum == null) {
                            sum = 0.0;
                        }
                        sum += event.getAmount();
                        sumState.update(sum);
                        
                        // 更新事件
                        event.setCount(count);
                        event.getProperties().put("sum", String.valueOf(sum));
                        event.getProperties().put("window-start", String.valueOf(windowStart));
                        
                        // 输出事件
                        out.collect(event);
                        
                        LOG.debug("Event processed in window successfully: {}", event.getId());
                    } catch (Exception e) {
                        LOG.error("Error processing event in window: " + event.getId(), e);
                    }
                }
            })
            .setParallelism(4);
    }
} 